package com.runa.monitor.platinterface.extend.nb.mqtt;

import com.alibaba.fastjson.JSONObject;
import com.ctiot.aep.mqmsgpush.sdk.IMsgListener;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.runa.monitor.platinterface.extend.nb.config.MqttNbIotConfig;
import com.runa.monitor.platinterface.extend.nb.config.NewMqttNbIotConfig;
import com.runa.monitor.platinterface.extend.nb.service.NbPlantService;
import com.runa.monitor.platinterface.extend.nb.utils.MqttSetUtils;
import com.runa.monitor.platinterface.extend.nb.utils.NewMqttSetUtils;
import com.runa.monitor.platinterface.extend.nb.vo.NewReportDataBaseVo;
import com.runa.monitor.platinterface.extend.nb.vo.ReportDataBaseVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.*;

@Slf4j
@Component
public class TopicMessageListener implements IMsgListener {


    @Resource
    private MqttNbIotConfig mqttNbIotConfig;

    @Resource
    private NewMqttNbIotConfig newMqttNbIotConfig;

    @Resource
    private NbPlantService mqttService;

    private static final ThreadFactory NAMED_THREAD_FACTORY = new ThreadFactoryBuilder()
            .setNameFormat("mqtt-topic-pool-%d").build();

    private static final ExecutorService newFixedThreadPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(),
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.AbortPolicy());


    @Override
    public void onMessage(String msg) {
        log.info("电信NBmqtt数据："+msg);
        String messageType = JSONObject.parseObject(msg).getString("messageType");
        String tenantId = JSONObject.parseObject(msg).getString("tenantId");
        if ("dataReport".equals(messageType)) {
            //接收消息
            log.info("---电信mqtt收到租户id：" + tenantId + ",NB-IoT设备上报的数据：" + msg);
            if (newMqttNbIotConfig.getEnable() && newMqttNbIotConfig.getTenantId().equals(tenantId)) {
                log.info("---小瑞面板tenantId：{}", newMqttNbIotConfig.getTenantId());
                NewReportDataBaseVo reportBaseDTO = JSONObject.parseObject(msg, NewReportDataBaseVo.class);
                newFixedThreadPool.execute(new NewMqttUpDataQueue(NewMqttSetUtils.APP_KEY, NewMqttSetUtils.APP_SECRET, mqttService, reportBaseDTO));
            } else if (mqttNbIotConfig.getEnable() && mqttNbIotConfig.getTenantId().equals(tenantId)) {
                ReportDataBaseVo reportBaseDTO = JSONObject.parseObject(msg, ReportDataBaseVo.class);
                newFixedThreadPool.execute(new MqttUpDataQueue(MqttSetUtils.APP_KEY, MqttSetUtils.APP_SECRET, mqttService, reportBaseDTO));
            }
        } else {
            log.info("电信无关数据---电信mqtt收到设备上数的数据：" + msg);
        }
    }
}
